package com.rrd.dw.mr.gzip.kafka;

import java.io.IOException;
import java.util.Date;
import java.util.Set;
import java.util.regex.Pattern;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.MD5Util;
import com.ucredit.babel.common.model.Job;
import com.ucredit.babel.common.model.dhb.DhbItagInfo;
import com.ucredit.babel.common.model.dhb.DhbItagInfoRecords;

/**
 * 互金数据
 * @author liubaoxin
 */
@Deprecated
public class DHBTagInfoParse {
	public static final String CRT001 =  "\001";
	public static final Pattern CRT001_PATTERN =  Pattern.compile("\001");
	private String keyStr="";
	private String timestamstr="";
	private String job_id ="";
	private String system_id ="";
	private StringBuilder sb = new StringBuilder(500);
	private StringBuilder sb2 = new StringBuilder(500);
	private  Text val = new Text();
	 private Text key_text = new Text("nifa_query");
     private Text key_deail_text = new Text("nifa_query_detail");
	private JSONObject dataObj = null;
	private JSONObject tagsObj = null;
	 private JSONObject tmpObj = null;
	private String tmpstr=null;
	 JSONArray catnames = null;
     JSONArray itagIds = null;
     JSONObject flagObject=null;
     
	/**
	 * 互金数据
	 * @param userBehavior
	 * @throws InterruptedException 
	 * @throws IOException 
	 */
	public  void parseNIFA(Context context,DhbItagInfoRecords tagRecord,String successful) throws IOException, InterruptedException{
		 
    	Job job = tagRecord.getJob();
  
    	DhbItagInfo tagInfo = tagRecord.getItagInfo();
    	if(tagInfo==null){
    		return;
    	}
    	sb.delete(0,sb.length());
    	//job_id
    	this.job_id = job.getJobID()==null?"":job.getJobID();
    	//system_id
    	this.system_id = job.getSystemID()==null?"":job.getSystemID().toString();
    	//数据时间 
    	this.timestamstr=tagRecord.getTimestamp()==null?"":DateUtil.formateDateStr(new Date(tagRecord.getTimestamp()));
    	 
    	keyStr = MD5Util.getMd5(job_id+system_id+timestamstr);
    	
    	sb.append(keyStr).append(CRT001).append(job_id).append(CRT001);
    	sb.append(job.isFinished()==null?"":job.isFinished()).append(CRT001);
    	sb.append(job.isRequested()==null?"":job.isRequested()).append(CRT001);
    	sb.append(job.isSuccessful()==null?"":job.isSuccessful()).append(CRT001);
    	sb.append(job.getStartTime()==null?"":DateUtil.formateDateStr(job.getStartTime())).append(CRT001);
    	sb.append(job.getEndTime()==null?"":DateUtil.formateDateStr(job.getEndTime())).append(CRT001);
    	sb.append(job.getDataFile()==null?"":job.getDataFile()).append(CRT001);
    	sb.append(system_id).append(CRT001);
    	sb.append(job.getQuery()==null?"":job.getQuery()).append(CRT001);
    	sb.append(job.getTryTimes()==null?"":job.getTryTimes()).append(CRT001);
    	sb.append(tagRecord.getType()==null?"":tagRecord.getType()).append(CRT001);
    	sb.append(tagInfo.getMessage()==null?"":tagInfo.getMessage()).append(CRT001);
    	sb.append(tagInfo.getCode()==null?"":tagInfo.getCode()).append(CRT001);
    	sb.append(timestamstr);
    	key_text.set("dhb_tag_job");
    	val.set(sb.toString());
    	context.write(key_text, val);
    	 
    	 
    	String data_json = tagInfo.getData();
    	if(data_json==null || data_json.trim().equals("") || data_json.equals("[]")){
    		return;
    	}
    	
    	String code = tagInfo.getCode()==null?"":tagInfo.getCode();
    	String message = tagInfo.getMessage()==null?"":tagInfo.getMessage();
    	String sid = tagInfo.getSid() == null?"":tagInfo.getSid();
    	 //解析data_json
        dataObj=JSON.parseObject(data_json);
        if(dataObj==null){
          return;
        }else{
          tagsObj = dataObj.getJSONObject("tags");
          if(tagsObj==null || tagsObj.size()<=0){
        	  return;
          }
          Set<String> tagSet = tagsObj.keySet();
         
          for (String tagstr : tagSet) {
              tmpObj = tagsObj.getJSONObject(tagstr);
              if (tmpObj == null)
                  continue;

              sb.delete(0, sb.length());
              sb.append(keyStr).append(CRT001).append(code).append(CRT001).append(message).append(CRT001);
              sb.append(sid).append(CRT001);
              sb.append(tagstr).append(CRT001);
              sb.append(tmpObj.getString("status")==null?"":tmpObj.getString("status")).append(CRT001);
              sb.append(tmpObj.getString("name")==null?"":tmpObj.getString("name")).append(CRT001);
              sb.append(tmpObj.getString("teldesc")==null?"":tmpObj.getString("teldesc")).append(CRT001);
              sb.append(tmpObj.getString("telnum")==null?"":tmpObj.getString("telnum")).append(CRT001);
              sb.append(tmpObj.getString("tellloc")==null?"":tmpObj.getString("tellloc")).append(CRT001);
              flagObject = tmpObj.getJSONObject("flag");
              sb.append(flagObject==null||flagObject.getString("type")==null?"":flagObject.getString("type")).append(CRT001);
              sb.append(flagObject==null||flagObject.getString("num")==null?"":flagObject.getString("num")).append(CRT001);
              sb.append(flagObject==null||flagObject.getString("fid")==null?"":flagObject.getString("fid"));
              key_text.set("dhb_tag_record");
          	  val.set(sb.toString());
          	  context.write(key_text, val);
          	catnames = tmpObj.getJSONArray("catnames");
          	if (catnames != null) {
                for (int i = 0; i < catnames.size(); i++) {
                    sb.delete(0, sb.length());
                    sb.append(keyStr).append(CRT001);
                    sb.append(sid).append(CRT001);
                    sb.append(tagstr).append(CRT001);
                    sb.append((i + 1)).append(CRT001);
                    sb.append(catnames.getString(i)); 
                    key_text.set("dhb_tag_catnames");
                    val.set(sb.toString());
                    context.write(key_text, val);
                }
            }
            itagIds = tmpObj.getJSONArray("itag_ids");
            sb.delete(0, sb.length());
            if (itagIds != null) {
                for (int i = 0; i < itagIds.size(); i++) {
                    sb.delete(0, sb.length());
                    sb.append(keyStr).append(CRT001);
                    sb.append(sid).append(CRT001);
                    sb.append(tagstr).append(CRT001);
                    sb.append((i + 1)).append(CRT001);
                    sb.append(itagIds.getString(i));
                    key_text.set("dhb_tag_ids");
                    val.set(sb.toString());
                    context.write(key_text, val);
                }
            }
          }
          
        }
    	 
	}
	
	
	
	
	
}
